/**********
This library is free software; you can redistribute it and/or modify it under
the terms of the GNU Lesser General Public License as published by the
Free Software Foundation; either version 3 of the License, or (at your
option) any later version. (See <http://www.gnu.org/copyleft/lesser.html>.)

This library is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for
more details.

You should have received a copy of the GNU Lesser General Public License
along with this library; if not, write to the Free Software Foundation, Inc.,
51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
**********/
// "liveMedia"
// Copyright (c) 1996-2018 Live Networks, Inc.  All rights reserved.
// An abstraction of a network interface used for RTP (or RTCP).
// (This allows the RTP-over-TCP hack (RFC 2326, section 10.12) to
// be implemented transparently.)
// Implementation

#include "RTPInterface.hh"
#include <GroupsockHelper.hh>
#include <stdio.h>
#ifndef _WIN32
#include <sys/uio.h>
#endif
#include "../../trace.h"
////////// Helper Functions - Definition //////////

// Helper routines and data structures, used to implement
// sending/receiving RTP/RTCP over a TCP socket:

// Reading RTP-over-TCP is implemented using two levels of hash tables.
// The top-level hash table maps TCP socket numbers to a
// "SocketDescriptor" that contains a hash table for each of the
// sub-channels that are reading from this socket.

static HashTable* socketHashTable(UsageEnvironment& env, Boolean createIfNotPresent = True) {
  _Tables* ourTables = _Tables::getOurTables(env, createIfNotPresent);
  if (ourTables == NULL) return NULL;

  if (ourTables->socketTable == NULL) {
    // Create a new socket number -> SocketDescriptor mapping table:
    ourTables->socketTable = HashTable::create(ONE_WORD_HASH_KEYS);
  }
  return (HashTable*)(ourTables->socketTable);
}

class SocketDescriptor {
public:
  SocketDescriptor(UsageEnvironment& env, int socketNum);
  virtual ~SocketDescriptor();

  void registerRTPInterface(unsigned char streamChannelId,
			    RTPInterface* rtpInterface);
  RTPInterface* lookupRTPInterface(unsigned char streamChannelId);
  void deregisterRTPInterface(unsigned char streamChannelId);

  void setServerRequestAlternativeByteHandler(ServerRequestAlternativeByteHandler* handler, void* clientData) {
    fServerRequestAlternativeByteHandler = handler;
    fServerRequestAlternativeByteHandlerClientData = clientData;
  }

private:
  static void tcpReadHandler(SocketDescriptor*, int mask);
  Boolean tcpReadHandler1(int mask);

private:
  UsageEnvironment& fEnv;
  int fOurSocketNum;
  HashTable* fSubChannelHashTable;
  ServerRequestAlternativeByteHandler* fServerRequestAlternativeByteHandler;
  void* fServerRequestAlternativeByteHandlerClientData;
  u_int8_t fStreamChannelId, fSizeByte1;
  Boolean fReadErrorOccurred, fDeleteMyselfNext, fAreInReadHandlerLoop;
  enum { AWAITING_DOLLAR, AWAITING_STREAM_CHANNEL_ID, AWAITING_SIZE1, AWAITING_SIZE2, AWAITING_PACKET_DATA } fTCPReadingState;
};

static SocketDescriptor* lookupSocketDescriptor(UsageEnvironment& env, int sockNum, Boolean createIfNotFound = True) {
  HashTable* table = socketHashTable(env, createIfNotFound);
  if (table == NULL) return NULL;

  char const* key = (char const*)(long)sockNum;
  SocketDescriptor* socketDescriptor = (SocketDescriptor*)(table->Lookup(key));
  if (socketDescriptor == NULL) {
    if (createIfNotFound) {

#ifdef _DEBUG
	  //_TRACE(TRACE_LOG_DEBUG, (char*)"socketDescriptor = new SocketDescriptor(env, sockNum); [%s]\n", env.GetEnvirName());
#endif

      socketDescriptor = new SocketDescriptor(env, sockNum);
      table->Add((char const*)(long)(sockNum), socketDescriptor);
    } else if (table->IsEmpty()) {
      // We can also delete the table (to reclaim space):
      _Tables* ourTables = _Tables::getOurTables(env);
      delete table;
      ourTables->socketTable = NULL;
      ourTables->reclaimIfPossible();
    }
  }

  return socketDescriptor;
}

static void removeSocketDescription(UsageEnvironment& env, int sockNum) {
  char const* key = (char const*)(long)sockNum;
  HashTable* table = socketHashTable(env);
  table->Remove(key);

  if (table->IsEmpty()) {
    // We can also delete the table (to reclaim space):
    _Tables* ourTables = _Tables::getOurTables(env);
    delete table;
    ourTables->socketTable = NULL;
    ourTables->reclaimIfPossible();
  }
}


////////// RTPInterface - Implementation //////////
#ifndef RTPINTERFACE_BLOCKING_WRITE_TIMEOUT_MS
#define RTPINTERFACE_BLOCKING_WRITE_TIMEOUT_MS 500
#endif

RTPInterface::RTPInterface(Medium* owner, Groupsock* gs)
  : fOwner(owner), fGS(gs),
    fTCPStreams(NULL),
    fNextTCPReadSize(0), fNextTCPReadStreamSocketNum(-1),
    fNextTCPReadStreamChannelId(0xFF), fReadHandlerProc(NULL),
    fAuxReadHandlerFunc(NULL), fAuxReadHandlerClientData(NULL) {
  // Make the socket non-blocking, even though it will be read from only asynchronously, when packets arrive.
  // The reason for this is that, in some OSs, reads on a blocking socket can (allegedly) sometimes block,
  // even if the socket was previously reported (e.g., by "select()") as having data available.
  // (This can supposedly happen if the UDP checksum fails, for example.)
#ifdef _WIN32
		//win32才启用异步
  makeSocketNonBlocking(fGS->socketNum());
#endif

  //对于同一路视频流，当请求的VLC客户端越来越多时，会发现后面请求的VLC客户端正在播放但没有图像。
  //我们找到RTPInterface.cpp文件中RTPInterface的构造函数，注释掉其中调用makeSokcetNonBlocking函数的那一句即可
  //increaseSendBufferTo(envir(), fGS->socketNum(), 50*1024);
  increaseSendBufferTo(envir(), fGS->socketNum(), 512*1024);

	mFlag = LIVE_FLAG;
}

RTPInterface::~RTPInterface() {
  stopNetworkReading();
  delete fTCPStreams;
  fTCPStreams = NULL;
  mFlag = 0;
}

void RTPInterface::setStreamSocket(int sockNum,
				   unsigned char streamChannelId) {
  fGS->removeAllDestinations();

  _TRACE(TRACE_LOG_DEBUG, (char*)"RTPInterface::setStreamSocket从[%s]中移除该Socket[%d].\n", envir().GetEnvirName(), fGS->socketNum());

  envir().taskScheduler().disableBackgroundHandling(fGS->socketNum()); // turn off any reading on our datagram socket
  fGS->reset(); // and close our datagram socket, because we won't be using it anymore

  addStreamSocket(sockNum, streamChannelId);
}

void RTPInterface::addStreamSocket(int sockNum,
				   unsigned char streamChannelId) {
  if (sockNum < 0) return;

  for (tcpStreamRecord* streams = fTCPStreams; streams != NULL;
       streams = streams->fNext) {
    if (streams->fStreamSocketNum == sockNum
	&& streams->fStreamChannelId == streamChannelId) {
      return; // we already have it
    }
  }

  fTCPStreams = new tcpStreamRecord(sockNum, streamChannelId, fTCPStreams);

  //_TRACE(TRACE_LOG_INFO, (char*)"[%s] RTPInterface::addStreamSocket sockNum[%d] streamPtr[0x%X] streamChannelId[%d]\n", envir().GetEnvirName(), sockNum, fTCPStreams, streamChannelId);

  // Also, make sure this new socket is set up for receiving RTP/RTCP-over-TCP:
  SocketDescriptor* socketDescriptor = lookupSocketDescriptor(envir(), sockNum);
  socketDescriptor->registerRTPInterface(streamChannelId, this);
}

static void deregisterSocket(UsageEnvironment& env, int sockNum, unsigned char streamChannelId) {
  SocketDescriptor* socketDescriptor = lookupSocketDescriptor(env, sockNum, False);
  if (socketDescriptor != NULL) {
    socketDescriptor->deregisterRTPInterface(streamChannelId);
        // Note: This may delete "socketDescriptor",
        // if no more interfaces are using this socket

	//2019.12.30
	//env.taskScheduler().turnOffBackgroundReadHandling(sockNum);
	//closeSocket(sockNum);
  }
}

void RTPInterface::removeStreamSocket(int sockNum,
				      unsigned char streamChannelId) {
  // Remove - from our list of 'TCP streams' - the record of the (sockNum,streamChannelId) pair.
  // (However "streamChannelId" == 0xFF is a special case, meaning remove all
  //  (sockNum,*) pairs.)
  
  while (1) {
    tcpStreamRecord** streamsPtr = &fTCPStreams;

    while (*streamsPtr != NULL) {
      if ((*streamsPtr)->fStreamSocketNum == sockNum
	  && (streamChannelId == 0xFF || streamChannelId == (*streamsPtr)->fStreamChannelId)) {
	// Delete the record pointed to by *streamsPtr :

	//_TRACE(TRACE_LOG_INFO, (char*)"[%s] RTPInterface::removeStreamSocket sockNum[%d] streamPtr[0x%X] streamChannelId[%d] BEGIN\n", 
	//		envir().GetEnvirName(), sockNum, *streamsPtr, (*streamsPtr)->fStreamChannelId);

	if ((*streamsPtr) && (*streamsPtr)->mFlag == LIVE_FLAG)
	{
		unsigned char streamChannelIdToRemove = (*streamsPtr)->fStreamChannelId;
		tcpStreamRecord* next = (*streamsPtr)->fNext;
		(*streamsPtr)->fNext = NULL;
		delete (*streamsPtr);
		*streamsPtr = next;

		// And 'deregister' this socket,channelId pair:
		deregisterSocket(envir(), sockNum, streamChannelIdToRemove);
	}

	if (streamChannelId != 0xFF) return; // we're done
	break; // start again from the beginning of the list, in case the list has changed
      } else {
	streamsPtr = &((*streamsPtr)->fNext);
      }
    }
    if (*streamsPtr == NULL) break;
  }
}

void RTPInterface::setServerRequestAlternativeByteHandler(UsageEnvironment& env, int socketNum,
							  ServerRequestAlternativeByteHandler* handler, void* clientData) {
  SocketDescriptor* socketDescriptor = lookupSocketDescriptor(env, socketNum, False);

  if (socketDescriptor != NULL) socketDescriptor->setServerRequestAlternativeByteHandler(handler, clientData);
}

void RTPInterface::clearServerRequestAlternativeByteHandler(UsageEnvironment& env, int socketNum) {
  setServerRequestAlternativeByteHandler(env, socketNum, NULL, NULL);
}




Boolean RTPInterface::sendRTPorRTCPPacketOverTCP(tcpStreamRecord* tcpStream, unsigned char* packet, unsigned packetSize)
{
	//if (streamChannelId != 0x00 && streamChannelId != 0x02)	return True;

	Boolean bRet = True;

	if (NULL == tcpStream)		return False;
	if (tcpStream->fStreamSocketNum < 1)		return False;

	int tcpSocket = tcpStream->fStreamSocketNum;
	unsigned char streamChannelId = tcpStream->fStreamChannelId;

	do 
	{
		u_int8_t framingHeader[4];
		framingHeader[0] = '$';
		framingHeader[1] = streamChannelId;
		framingHeader[2] = (u_int8_t) ((packetSize&0xFF00)>>8);
		framingHeader[3] = (u_int8_t) (packetSize&0xFF);
#ifndef _WIN32
		if (tcpStream->fRTPPacket[0].bufpos + 4 + (int)packetSize <= tcpStream->fRTPPacket[0].bufsize)
		{
			memcpy(tcpStream->fRTPPacket[0].pbuf+tcpStream->fRTPPacket[0].bufpos, &framingHeader[0], 4);		tcpStream->fRTPPacket[0].bufpos += 4;
			memcpy(tcpStream->fRTPPacket[0].pbuf+tcpStream->fRTPPacket[0].bufpos, packet, packetSize);		tcpStream->fRTPPacket[0].bufpos += packetSize;

			tcpStream->fRTPPacket[0].packetNum ++;
		}
		else if (tcpStream->fRTPPacket[1].bufpos + 4 + (int)packetSize <= tcpStream->fRTPPacket[1].bufsize)
		{
			memcpy(tcpStream->fRTPPacket[1].pbuf+tcpStream->fRTPPacket[1].bufpos, &framingHeader[0], 4);		tcpStream->fRTPPacket[1].bufpos += 4;
			memcpy(tcpStream->fRTPPacket[1].pbuf+tcpStream->fRTPPacket[1].bufpos, packet, packetSize);		tcpStream->fRTPPacket[1].bufpos += packetSize;

			tcpStream->fRTPPacket[1].packetNum ++;
		}
		else
		{
			printf("Video rtp packet .....\n");
		}

		if ( (tcpStream->fRTPPacket[0].packetNum + tcpStream->fRTPPacket[1].packetNum >= 8) || ((streamChannelId == 0) && (packetSize<1456)) ||
			(streamChannelId > 0) )
			//(streamChannelId==1 || streamChannelId==2 || streamChannelId==3)  )
		{
	#if 0
			if (tcpStream->fRTPPacket[0].packetNum + tcpStream->fRTPPacket[1].packetNum < 8)
			{
			printf("packetnum:%d packetsize:%d, channelid:%d\n", tcpStream->fRTPPacket[0].packetNum + tcpStream->fRTPPacket[1].packetNum, packetSize, streamChannelId);
			}
	#endif

			bRet = sendDataOverTCPEx(tcpSocket, tcpStream->fRTPPacket[0].pbuf, tcpStream->fRTPPacket[0].bufpos, tcpStream->fRTPPacket[1].pbuf, tcpStream->fRTPPacket[1].bufpos, 0);

			for (int i=0; i<MAX_RTP_PACKET_NUM; i++)
			{
				tcpStream->fRTPPacket[i].packetNum = 0;
				tcpStream->fRTPPacket[i].bufpos = 0;
			}
		}
#else
	bRet = sendDataOverTCPEx(tcpSocket, framingHeader, 4, packet, packetSize, 0);
#endif

#ifdef DEBUG_SEND
    _TRACE(TRACE_LOG_ERROR, (char*)"sendRTPorRTCPPacketOverTCP: completed\n"); fflush(stderr);
#endif

    return bRet;
  } while (0);

#ifdef DEBUG_SEND
  _TRACE(TRACE_LOG_ERROR, (char*)"sendRTPorRTCPPacketOverTCP: failed! (errno %d)\n", envir().getErrno()); fflush(stderr);
#endif
  return bRet;
}

Boolean RTPInterface::sendPacket(unsigned char* packet, unsigned packetSize) {
  Boolean success = True; // we'll return False instead if any of the sends fail

  if (GetFlag() != LIVE_FLAG)			return False;

  // Normal case: Send as a UDP packet:
  if (!fGS->output(envir(), packet, packetSize)) success = False;

#ifdef _DEBUG1
  _TRACE(TRACE_LOG_DEBUG, (char*)"sendPacket.  [%s]\n", envir().GetEnvirName());
#endif

  // Also, send over each of our TCP sockets:
  tcpStreamRecord* nextStream;
  for (tcpStreamRecord* stream = fTCPStreams; stream != NULL; stream = nextStream) {
    nextStream = stream->fNext; // Set this now, in case the following deletes "stream":
	
	if (! sendRTPorRTCPPacketOverTCP(stream, packet, packetSize))
	{
		success = False;
	}
  }

  return success;
}

void RTPInterface
::startNetworkReading(TaskScheduler::BackgroundHandlerProc* handlerProc) {
  // Normal case: Arrange to read UDP packets:
  envir().taskScheduler().
    turnOnBackgroundReadHandling(fGS->socketNum(), handlerProc, fOwner);

  // Also, receive RTP over TCP, on each of our TCP connections:
  fReadHandlerProc = handlerProc;
  for (tcpStreamRecord* streams = fTCPStreams; streams != NULL;
       streams = streams->fNext) {
    // Get a socket descriptor for "streams->fStreamSocketNum":
    SocketDescriptor* socketDescriptor = lookupSocketDescriptor(envir(), streams->fStreamSocketNum);

    // Tell it about our subChannel:
    socketDescriptor->registerRTPInterface(streams->fStreamChannelId, this);
  }
}

Boolean RTPInterface::handleRead(unsigned char* buffer, unsigned bufferMaxSize,
				 unsigned& bytesRead, LIVE_NET_ADDRESS_SOCKADDR_IN& fromAddress,
				 int& tcpSocketNum, unsigned char& tcpStreamChannelId,
				 Boolean& packetReadWasIncomplete) {
  packetReadWasIncomplete = False; // by default
  Boolean readSuccess;
  if (fNextTCPReadStreamSocketNum < 0) {
    // Normal case: read from the (datagram) 'groupsock':
    tcpSocketNum = -1;
    readSuccess = fGS->handleRead(buffer, bufferMaxSize, bytesRead, fromAddress);
  } else {
    // Read from the TCP connection:
    tcpSocketNum = fNextTCPReadStreamSocketNum;
    tcpStreamChannelId = fNextTCPReadStreamChannelId;

    bytesRead = 0;
    unsigned totBytesToRead = fNextTCPReadSize;
    if (totBytesToRead > bufferMaxSize) totBytesToRead = bufferMaxSize;
    unsigned curBytesToRead = totBytesToRead;
    int curBytesRead;
    while ((curBytesRead = readSocket(envir(), fNextTCPReadStreamSocketNum,
				      &buffer[bytesRead], curBytesToRead,
				      fromAddress)) > 0) {
      bytesRead += curBytesRead;
      if (bytesRead >= totBytesToRead) break;
      curBytesToRead -= curBytesRead;
    }
    fNextTCPReadSize -= bytesRead;
    if (fNextTCPReadSize == 0) {
      // We've read all of the data that we asked for
      readSuccess = True;
    } else if (curBytesRead < 0) {
      // There was an error reading the socket
      bytesRead = 0;
      readSuccess = False;
    } else {
      // We need to read more bytes, and there was not an error reading the socket
      packetReadWasIncomplete = True;
      return True;
    }
    fNextTCPReadStreamSocketNum = -1; // default, for next time
  }

  if (readSuccess && fAuxReadHandlerFunc != NULL) {
    // Also pass the newly-read packet data to our auxilliary handler:
    (*fAuxReadHandlerFunc)(fAuxReadHandlerClientData, buffer, bytesRead);
  }
  return readSuccess;
}

void RTPInterface::stopNetworkReading() {
  // Normal case
  if (fGS != NULL) envir().taskScheduler().turnOffBackgroundReadHandling(fGS->socketNum());

  // Also turn off read handling on each of our TCP connections:
  for (tcpStreamRecord* streams = fTCPStreams; streams != NULL; streams = streams->fNext) {
    deregisterSocket(envir(), streams->fStreamSocketNum, streams->fStreamChannelId);
  }
}


////////// Helper Functions - Implementation /////////

Boolean RTPInterface::sendDataOverTCPEx(int socketNum, u_int8_t * data1, unsigned dataSize1, u_int8_t * data2, unsigned dataSize2, int level) {

#ifndef _WIN32
	Boolean bRet = False;

	struct iovec iov[2];
	iov[0].iov_base = data1;
	iov[0].iov_len = dataSize1;
	iov[1].iov_base = data2;
	iov[1].iov_len = dataSize2;
	int write_bytes = writev(socketNum, iov, (dataSize2>1?2:1));

	if (write_bytes < (int)(dataSize1 + dataSize2) )
	{
		if (write_bytes < 0 && envir().getErrno() != EAGAIN) {
		//if (write_bytes < 1){// && envir().getErrno() != EAGAIN) {
		  // Because the "send()" call failed, assume that the socket is now unusable, so stop
		  // using it (for both RTP and RTCP):
		  removeStreamSocket(socketNum, 0xFF);
		  return False;
		}
		
		if (level == 0x00)
		{
      		makeSocketBlocking(socketNum, RTPINTERFACE_BLOCKING_WRITE_TIMEOUT_MS);
		}		
		
		if (write_bytes>0 && write_bytes < dataSize1)
		{
			printf("write_bytes<dataSize1: %d / %d\n", write_bytes, dataSize1+dataSize2);
			bRet = sendDataOverTCPEx(socketNum, data1+write_bytes, dataSize1-write_bytes, data2, dataSize2, level++);
		}
		else if (write_bytes>=dataSize1)
		{
			printf("write_bytes>dataSize1: %d / %d\n", write_bytes, dataSize1+dataSize2);
			bRet = sendDataOverTCPEx(socketNum, data2+write_bytes-dataSize1, dataSize2+dataSize1-write_bytes, NULL, 0, level++);
		}

		if (level == 0x00)
		{
			makeSocketNonBlocking(socketNum);
		}
		
		return bRet;
	}
	else
	{
		return True;
	}

#else

	do
	{
		if (! sendDataOverTCP(socketNum, data1, dataSize1, False)) break;

		if (dataSize2 > 0)
		{
			if (! sendDataOverTCP(socketNum, data2, dataSize2, True)) break;
		}

		return True;
	}while(0);
#endif
	return False;
}


#ifndef RTPINTERFACE_BLOCKING_WRITE_TIMEOUT_MS
#define RTPINTERFACE_BLOCKING_WRITE_TIMEOUT_MS 500
#endif

Boolean RTPInterface::sendDataOverTCP(int socketNum, u_int8_t const* data, unsigned dataSize, Boolean forceSendToSucceed) {
  int sendResult = send(socketNum, (char const*)data, dataSize, 0/*flags*/);
  if (sendResult < (int)dataSize) {
    // The TCP send() failed - at least partially.

    unsigned numBytesSentSoFar = sendResult < 0 ? 0 : (unsigned)sendResult;
    if (numBytesSentSoFar > 0 || (forceSendToSucceed && envir().getErrno() == EAGAIN)) {
      // The OS's TCP send buffer has filled up (because the stream's bitrate has exceeded
      // the capacity of the TCP connection!).
      // Force this data write to succeed, by blocking if necessary until it does:
      unsigned numBytesRemainingToSend = dataSize - numBytesSentSoFar;
#ifdef DEBUG_SEND
      fprintf(stderr, "sendDataOverTCP: resending %d-byte send (blocking)\n", numBytesRemainingToSend); fflush(stderr);
#endif
      makeSocketBlocking(socketNum, RTPINTERFACE_BLOCKING_WRITE_TIMEOUT_MS);
      sendResult = send(socketNum, (char const*)(&data[numBytesSentSoFar]), numBytesRemainingToSend, 0/*flags*/);
      if ((unsigned)sendResult != numBytesRemainingToSend) {
	// The blocking "send()" failed, or timed out.  In either case, we assume that the
	// TCP connection has failed (or is 'hanging' indefinitely), and we stop using it
	// (for both RTP and RTP).
	// (If we kept using the socket here, the RTP or RTCP packet write would be in an
	//  incomplete, inconsistent state.)
#ifdef DEBUG_SEND
	fprintf(stderr, "sendDataOverTCP: blocking send() failed (delivering %d bytes out of %d); closing socket %d\n", sendResult, numBytesRemainingToSend, socketNum); fflush(stderr);
#endif
	removeStreamSocket(socketNum, 0xFF);
	return False;
      }
      makeSocketNonBlocking(socketNum);

      return True;
    } else if (sendResult < 0 && envir().getErrno() != EAGAIN) {
      // Because the "send()" call failed, assume that the socket is now unusable, so stop
      // using it (for both RTP and RTCP):
      removeStreamSocket(socketNum, 0xFF);
    }

    return False;
  }

  return True;
}

SocketDescriptor::SocketDescriptor(UsageEnvironment& env, int socketNum)
  :fEnv(env), fOurSocketNum(socketNum),
    fSubChannelHashTable(HashTable::create(ONE_WORD_HASH_KEYS)),
   fServerRequestAlternativeByteHandler(NULL), fServerRequestAlternativeByteHandlerClientData(NULL),
   fReadErrorOccurred(False), fDeleteMyselfNext(False), fAreInReadHandlerLoop(False), fTCPReadingState(AWAITING_DOLLAR) {
}

SocketDescriptor::~SocketDescriptor() {
  fEnv.taskScheduler().turnOffBackgroundReadHandling(fOurSocketNum);
  removeSocketDescription(fEnv, fOurSocketNum);

  if (fSubChannelHashTable != NULL) {
    // Remove knowledge of this socket from any "RTPInterface"s that are using it:
    HashTable::Iterator* iter = HashTable::Iterator::create(*fSubChannelHashTable);
    RTPInterface* rtpInterface;
    char const* key;

    while ((rtpInterface = (RTPInterface*)(iter->next(key))) != NULL) {
      u_int64_t streamChannelIdLong = (u_int64_t)key;
      unsigned char streamChannelId = (unsigned char)streamChannelIdLong;

	  if (rtpInterface->GetFlag() == LIVE_FLAG)
	  {
		rtpInterface->removeStreamSocket(fOurSocketNum, streamChannelId);
	  }
    }
    delete iter;

    // Then remove the hash table entries themselves, and then remove the hash table:
    while (fSubChannelHashTable->RemoveNext() != NULL) {}
    delete fSubChannelHashTable;
	fSubChannelHashTable = NULL;
  }

  // Finally:
  if (fServerRequestAlternativeByteHandler != NULL) {
    // Hack: Pass a special character to our alternative byte handler, to tell it that either
    // - an error occurred when reading the TCP socket, or
    // - no error occurred, but it needs to take over control of the TCP socket once again.
    u_int8_t specialChar = fReadErrorOccurred ? 0xFF : 0xFE;
    (*fServerRequestAlternativeByteHandler)(fServerRequestAlternativeByteHandlerClientData, specialChar);
  }
}

void SocketDescriptor::registerRTPInterface(unsigned char streamChannelId,
					    RTPInterface* rtpInterface) {
  Boolean isFirstRegistration = fSubChannelHashTable->IsEmpty();
#if defined(DEBUG_SEND)||defined(DEBUG_RECEIVE)
  fprintf(stderr, "SocketDescriptor(socket %d)::registerRTPInterface(channel %d): isFirstRegistration %d\n", fOurSocketNum, streamChannelId, isFirstRegistration);
#endif
  fSubChannelHashTable->Add((char const*)(long)streamChannelId,
			    rtpInterface);

  if (isFirstRegistration) {

#ifdef _DEBUG
	  //_TRACE(TRACE_LOG_DEBUG, (char*)"SocketDescriptor::registerRTPInterface [%s]\n", fEnv.GetEnvirName());
#endif

    // Arrange to handle reads on this TCP socket:
    TaskScheduler::BackgroundHandlerProc* handler
      = (TaskScheduler::BackgroundHandlerProc*)&tcpReadHandler;
    fEnv.taskScheduler().
      setBackgroundHandling(fOurSocketNum, SOCKET_READABLE|SOCKET_EXCEPTION, handler, this);
  }
}

RTPInterface* SocketDescriptor
::lookupRTPInterface(unsigned char streamChannelId) {
  char const* lookupArg = (char const*)(long)streamChannelId;
  return (RTPInterface*)(fSubChannelHashTable->Lookup(lookupArg));
}

void SocketDescriptor
::deregisterRTPInterface(unsigned char streamChannelId) {
#if defined(DEBUG_SEND)||defined(DEBUG_RECEIVE)
  fprintf(stderr, "SocketDescriptor(socket %d)::deregisterRTPInterface(channel %d)\n", fOurSocketNum, streamChannelId);
#endif
  fSubChannelHashTable->Remove((char const*)(long)streamChannelId);

  if (fSubChannelHashTable->IsEmpty()) {
    // No more interfaces are using us, so it's curtains for us now:
    if (fAreInReadHandlerLoop) {
      fDeleteMyselfNext = True; // we can't delete ourself yet, but we'll do so from "tcpReadHandler()" below
    } else {
      delete this;
    }
  }
}

void SocketDescriptor::tcpReadHandler(SocketDescriptor* socketDescriptor, int mask) {
  // Call the read handler until it returns false, with a limit to avoid starving other sockets
  unsigned count = 2000;
  socketDescriptor->fAreInReadHandlerLoop = True;
  while (!socketDescriptor->fDeleteMyselfNext && socketDescriptor->tcpReadHandler1(mask) && --count > 0) {}
  socketDescriptor->fAreInReadHandlerLoop = False;
  if (socketDescriptor->fDeleteMyselfNext) delete socketDescriptor;
}

Boolean SocketDescriptor::tcpReadHandler1(int mask) {
  // We expect the following data over the TCP channel:
  //   optional RTSP command or response bytes (before the first '$' character)
  //   a '$' character
  //   a 1-byte channel id
  //   a 2-byte packet size (in network byte order)
  //   the packet data.
  // However, because the socket is being read asynchronously, this data might arrive in pieces.
  
  u_int8_t c;
  LIVE_NET_ADDRESS_SOCKADDR_IN fromAddress;
  if (fTCPReadingState != AWAITING_PACKET_DATA) {
    int result = readSocket(fEnv, fOurSocketNum, &c, 1, fromAddress);
    if (result == 0) { // There was no more data to read
      return False;
    } else if (result != 1) { // error reading TCP socket, so we will no longer handle it
#ifdef DEBUG_RECEIVE
      fprintf(stderr, "SocketDescriptor(socket %d)::tcpReadHandler(): readSocket(1 byte) returned %d (error)\n", fOurSocketNum, result);
#endif
      fReadErrorOccurred = True;
      fDeleteMyselfNext = True;
      return False;
    }
  }

  Boolean callAgain = True;
  switch (fTCPReadingState) {
    case AWAITING_DOLLAR: {
      if (c == '$') {
#ifdef DEBUG_RECEIVE
	fprintf(stderr, "SocketDescriptor(socket %d)::tcpReadHandler(): Saw '$'\n", fOurSocketNum);
#endif
	fTCPReadingState = AWAITING_STREAM_CHANNEL_ID;
      } else {
	// This character is part of a RTSP request or command, which is handled separately:
	if (fServerRequestAlternativeByteHandler != NULL && c != 0xFF && c != 0xFE) {
	  // Hack: 0xFF and 0xFE are used as special signaling characters, so don't send them
	  (*fServerRequestAlternativeByteHandler)(fServerRequestAlternativeByteHandlerClientData, c);
	}
      }
      break;
    }
    case AWAITING_STREAM_CHANNEL_ID: {
      // The byte that we read is the stream channel id.
      if (lookupRTPInterface(c) != NULL) { // sanity check
	fStreamChannelId = c;
	fTCPReadingState = AWAITING_SIZE1;
      } else {
	// This wasn't a stream channel id that we expected.  We're (somehow) in a strange state.  Try to recover:
#ifdef DEBUG_RECEIVE
	fprintf(stderr, "SocketDescriptor(socket %d)::tcpReadHandler(): Saw nonexistent stream channel id: 0x%02x\n", fOurSocketNum, c);
#endif
	fTCPReadingState = AWAITING_DOLLAR;
      }
      break;
    }
    case AWAITING_SIZE1: {
      // The byte that we read is the first (high) byte of the 16-bit RTP or RTCP packet 'size'.
      fSizeByte1 = c;
      fTCPReadingState = AWAITING_SIZE2;
      break;
    }
    case AWAITING_SIZE2: {
      // The byte that we read is the second (low) byte of the 16-bit RTP or RTCP packet 'size'.
      unsigned short size = (fSizeByte1<<8)|c;
      
      // Record the information about the packet data that will be read next:
      RTPInterface* rtpInterface = lookupRTPInterface(fStreamChannelId);
      if (rtpInterface != NULL) {
	rtpInterface->fNextTCPReadSize = size;
	rtpInterface->fNextTCPReadStreamSocketNum = fOurSocketNum;
	rtpInterface->fNextTCPReadStreamChannelId = fStreamChannelId;
      }
      fTCPReadingState = AWAITING_PACKET_DATA;
      break;
    }
    case AWAITING_PACKET_DATA: {
      callAgain = False;
      fTCPReadingState = AWAITING_DOLLAR; // the next state, unless we end up having to read more data in the current state
      // Call the appropriate read handler to get the packet data from the TCP stream:
      RTPInterface* rtpInterface = lookupRTPInterface(fStreamChannelId);
      if (rtpInterface != NULL) {
	if (rtpInterface->fNextTCPReadSize == 0) {
	  // We've already read all the data for this packet.
	  break;
	}
	if (rtpInterface->fReadHandlerProc != NULL) {
#ifdef DEBUG_RECEIVE
	  fprintf(stderr, "SocketDescriptor(socket %d)::tcpReadHandler(): reading %d bytes on channel %d\n", fOurSocketNum, rtpInterface->fNextTCPReadSize, rtpInterface->fNextTCPReadStreamChannelId);
#endif
	  fTCPReadingState = AWAITING_PACKET_DATA;
	  rtpInterface->fReadHandlerProc(rtpInterface->fOwner, mask);
	} else {
#ifdef DEBUG_RECEIVE
	  fprintf(stderr, "SocketDescriptor(socket %d)::tcpReadHandler(): No handler proc for \"rtpInterface\" for channel %d; need to skip %d remaining bytes\n", fOurSocketNum, fStreamChannelId, rtpInterface->fNextTCPReadSize);
#endif
	  int result = readSocket(fEnv, fOurSocketNum, &c, 1, fromAddress);
	  if (result < 0) { // error reading TCP socket, so we will no longer handle it
#ifdef DEBUG_RECEIVE
	    fprintf(stderr, "SocketDescriptor(socket %d)::tcpReadHandler(): readSocket(1 byte) returned %d (error)\n", fOurSocketNum, result);
#endif
	    fReadErrorOccurred = True;
	    fDeleteMyselfNext = True;
	    return False;
	  } else {
	    fTCPReadingState = AWAITING_PACKET_DATA;
	    if (result == 1) {
	      --rtpInterface->fNextTCPReadSize;
	      callAgain = True;
	    }
	  }
	}
      }
#ifdef DEBUG_RECEIVE
      else fprintf(stderr, "SocketDescriptor(socket %d)::tcpReadHandler(): No \"rtpInterface\" for channel %d\n", fOurSocketNum, fStreamChannelId);
#endif
    }
  }

  return callAgain;
}


////////// tcpStreamRecord implementation //////////

tcpStreamRecord
::tcpStreamRecord(int streamSocketNum, unsigned char streamChannelId,
		  tcpStreamRecord* next)
  : fNext(next),
    fStreamSocketNum(streamSocketNum), fStreamChannelId(streamChannelId), mFlag(LIVE_FLAG) {

	memset(&fRTPPacket[0], 0x00, sizeof(RTP_PACKE_T)*MAX_RTP_PACKET_NUM);
    int buf_size = 1460 * 6;
	for (int i=0; i<MAX_RTP_PACKET_NUM; i++)
	{
		fRTPPacket[i].pbuf = new u_int8_t[buf_size];
		if (NULL != fRTPPacket[i].pbuf)
		{
			fRTPPacket[i].bufsize = buf_size;
			memset(fRTPPacket[i].pbuf, 0x00, fRTPPacket[i].bufsize);
			fRTPPacket[i].bufpos = 0;
		}
	}
}

tcpStreamRecord::~tcpStreamRecord() {

	for (int i=0; i<MAX_RTP_PACKET_NUM; i++)
	{
		if (NULL != fRTPPacket[i].pbuf)
		{
			delete []fRTPPacket[i].pbuf;
			fRTPPacket[i].pbuf = NULL;
		}
	}
	memset(&fRTPPacket[0], 0x00, sizeof(RTP_PACKE_T)*MAX_RTP_PACKET_NUM);

	mFlag = 0;
  delete fNext;
}
